

/* -------------------------------------------------------------------------
 *
 * pgstat_replslot.c
 *	  复制槽统计的实现。
 *
 * 本文件包含复制槽统计的实现。它与pgstat.c保持分开，以便区分统计访问/
 * 存储实现与个别统计类型的详细信息。
 *
 * 复制槽统计与其他可变数量的统计工作略有不同。槽没有OIDs（因此可以在物理副本上创建）。
 * 在运行时使用槽索引作为对象ID。然而，当重新启动时，槽索引可能会改变。通过在（反）序列化时
 * 使用名称来解决这个问题。重新启动后，槽可能在关闭时被删除，这通过在启动时不恢复无法通过名
 * 称找到的槽的统计来解决。
 *
 * Copyright (c) 2001-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/utils/activity/pgstat_replslot.c
 * -------------------------------------------------------------------------
 */

#include "postgres.h"

#include "replication/slot.h"
#include "utils/builtins.h"		/* for namestrcpy() */
#include "utils/pgstat_internal.h"


static int	fc_get_replslot_index(const char *fc_name);


/*
 * 重置单个复制槽的计数器。
 *
 * 此功能的权限检查通过正常的
 * GRANT 系统进行管理。
 */
void pgstat_reset_replslot(const char *fc_name)
{
	ReplicationSlot *fc_slot;

	AssertArg(fc_name != NULL);

	/* 检查给定名称的槽是否存在。 */
	fc_slot = SearchNamedReplicationSlot(fc_name, true);

	if (!fc_slot)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("replication slot \"%s\" does not exist",
						fc_name)));

	/*
	 * 对于物理槽无需做任何操作，因为我们只收集逻辑
	 * 槽的统计信息。
	 */
	if (SlotIsPhysical(fc_slot))
		return;

	/* 重置这一条目 */
	pgstat_reset(PGSTAT_KIND_REPLSLOT, InvalidOid,
				 ReplicationSlotIndex(fc_slot));
}

/*
 * 报告复制槽统计信息。
 *
 * 我们可以依赖于槽的统计信息存在并属于此
 * 槽。只有在调用 pgstat_create_replslot() 或
 * pgstat_acquire_replslot() 后，我们才能到达这里。
 */
void pgstat_report_replslot(ReplicationSlot *fc_slot, const PgStat_StatReplSlotEntry *fc_repSlotStat)
{
	PgStat_EntryRef *fc_entry_ref;
	PgStatShared_ReplSlot *fc_shstatent;
	PgStat_StatReplSlotEntry *fc_statent;

	fc_entry_ref = pgstat_get_entry_ref_locked(PGSTAT_KIND_REPLSLOT, InvalidOid,
											ReplicationSlotIndex(fc_slot), false);
	fc_shstatent = (PgStatShared_ReplSlot *) fc_entry_ref->shared_stats;
	fc_statent = &fc_shstatent->stats;

	/* 更新复制槽统计信息 */
#define REPLSLOT_ACC(fld) fc_statent->fld += fc_repSlotStat->fld
	REPLSLOT_ACC(spill_txns);
	REPLSLOT_ACC(spill_count);
	REPLSLOT_ACC(spill_bytes);
	REPLSLOT_ACC(stream_txns);
	REPLSLOT_ACC(stream_count);
	REPLSLOT_ACC(stream_bytes);
	REPLSLOT_ACC(total_txns);
	REPLSLOT_ACC(total_bytes);
#undef REPLSLOT_ACC

	pgstat_unlock_entry(fc_entry_ref);
}

/*
 * 报告复制槽的创建。
 *
 * 注意：在已经持有 ReplicationSlotAllocationLock 时调用此函数，请
 * 小心调用回 slot.c 。
 */
void pgstat_create_replslot(ReplicationSlot *fc_slot)
{
	PgStat_EntryRef *fc_entry_ref;
	PgStatShared_ReplSlot *fc_shstatent;

	fc_entry_ref = pgstat_get_entry_ref_locked(PGSTAT_KIND_REPLSLOT, InvalidOid,
											ReplicationSlotIndex(fc_slot), false);
	fc_shstatent = (PgStatShared_ReplSlot *) fc_entry_ref->shared_stats;

	/*
	 * 注意：需要接受可能存在来自旧槽的统计信息，例如
	 * 如果我们在删除槽后之前崩溃。
	 */
	memset(&fc_shstatent->stats, 0, sizeof(fc_shstatent->stats));

	pgstat_unlock_entry(fc_entry_ref);
}

/*
 * 报告复制槽已被获取。
 *
 * 这保证了在后续的
 * pgstat_report_replslot() 调用期间存在统计条目。
 *
 * 如果我们之前崩溃，则没有统计数据存在。但如果我们没有崩溃，
 * 统计数据确实属于此槽：
 * - 统计不能属于已删除的槽，pgstat_drop_replslot() 会
 *   已被调用
 * - 如果槽在关闭期间被移除，
 *   pgstat_replslot_from_serialized_name_cb() 返回 false 会导致
 *   统计被丢弃
 */
void pgstat_acquire_replslot(ReplicationSlot *fc_slot)
{
	pgstat_get_entry_ref(PGSTAT_KIND_REPLSLOT, InvalidOid,
						 ReplicationSlotIndex(fc_slot), true, NULL);
}

/*
 * 报告复制槽的删除。
 */
void pgstat_drop_replslot(ReplicationSlot *fc_slot)
{
	if (!pgstat_drop_entry(PGSTAT_KIND_REPLSLOT, InvalidOid,
						   ReplicationSlotIndex(fc_slot)))
		pgstat_request_entry_refs_gc();
}

/*
 * SQL 可调用 pgstat* 函数的支持函数。返回
 * 指向复制槽统计结构的指针。
 */
PgStat_StatReplSlotEntry *
pgstat_fetch_replslot(NameData fc_slotname)
{
	int			fc_idx = fc_get_replslot_index(NameStr(fc_slotname));

	if (fc_idx == -1)
		return NULL;

	return (PgStat_StatReplSlotEntry *)
		pgstat_fetch_entry(PGSTAT_KIND_REPLSLOT, InvalidOid, fc_idx);
}

void pgstat_replslot_to_serialized_name_cb(const PgStat_HashKey *fc_key, const PgStatShared_Common *fc_header, NameData *fc_name)
{
	/*
	 * 这仅在关闭期间较晚调用。此时不允许更改现有槽的集合，
	 * 我们可以假设在偏移处存在一个槽。
	 */
	if (!ReplicationSlotName(fc_key->objoid, fc_name))
		elog(ERROR, "could not find name for replication slot index %u",
			 fc_key->objoid);
}

bool pgstat_replslot_from_serialized_name_cb(const NameData *fc_name, PgStat_HashKey *fc_key)
{
	int			fc_idx = fc_get_replslot_index(NameStr(*fc_name));

	/* 槽可能已被删除 */
	if (fc_idx == -1)
		return false;

	fc_key->kind = PGSTAT_KIND_REPLSLOT;
	fc_key->dboid = InvalidOid;
	fc_key->objoid = fc_idx;

	return true;
}

void pgstat_replslot_reset_timestamp_cb(PgStatShared_Common *fc_header, TimestampTz fc_ts)
{
	((PgStatShared_ReplSlot *) fc_header)->stats.stat_reset_timestamp = fc_ts;
}

static int fc_get_replslot_index(const char *fc_name)
{
	ReplicationSlot *fc_slot;

	AssertArg(fc_name != NULL);

	fc_slot = SearchNamedReplicationSlot(fc_name, true);

	if (!fc_slot)
		return -1;

	return ReplicationSlotIndex(fc_slot);
}
